package ru.ifmo.croak.impl.storage.team1;

import org.h2.fulltext.FullText;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.ifmo.croak.api.*;

import java.sql.*;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;


/**
 * @author Alex Vikharev
 *         21.05.11
 */
public class StorageImpl implements PostService, QueryService {
    private static final Logger log = LoggerFactory.getLogger(StorageImpl.class);
    private static final String CONNECTION_STRING = "jdbc:h2:";
    private static final int AUTHOR_NAME_MAX_LENGTH = 100;
    private static final int MESSAGE_MAX_LENGTH = 10000;

    private static final String CREATE_TABLE_QUERY = "create table if not exists MESSAGES " +
            " (ID BIGINT AUTO_INCREMENT PRIMARY KEY," +
            " UUID UUID NOT NULL UNIQUE," +
            " TIME BIGINT NOT NULL," +
            " AUTHOR VARCHAR(" + AUTHOR_NAME_MAX_LENGTH + ") NOT NULL," +
            " TEXT VARCHAR(" + MESSAGE_MAX_LENGTH + ") NOT NULL" +
            "  )";

    private static final String CHECK_INDEX_QUERY = "select count(*) from FT.INDEXES where SCHEMA='PUBLIC' and TABLE='MESSAGES' and COLUMNS='TEXT'";
    private static final String INSERT_MESSAGE_QUERY = "insert into MESSAGES (UUID, TIME, AUTHOR, TEXT) values (?, ?, ?, ?)";
    private static final int DUPLICATE_UUID_SQL_ERROR_CODE = 23505;

    private final Connection connection;
    private final Executor executor;
    final List<ListenerContainer> listeners;

    public Connection getConnection() {
        return connection;
    }

    private boolean isIndexCreated(Statement statement) throws SQLException {
        ResultSet rs = statement.executeQuery(CHECK_INDEX_QUERY);
        rs.next();
        return rs.getInt(1) > 0;
    }

    public StorageImpl(final String name) throws ClassNotFoundException, SQLException {
        log.info("Creating storage implementation with name: {}", name);
        Class.forName("org.h2.Driver");
        connection = DriverManager.getConnection(CONNECTION_STRING + name, "sa", "");

        Statement statement = connection.createStatement();
        statement.executeUpdate(CREATE_TABLE_QUERY);
        FullText.init(connection);
        if (!isIndexCreated(statement)) {
            log.info("FullText search tables doesn't exists. Creating new index.");
            FullText.createIndex(connection, "PUBLIC", "MESSAGES", "TEXT");
            log.info("FullText search index created.");
        } else {
            log.info("FullText search tables exists. Start reindexing");
            FullText.reindex(connection);
            log.info("Reindexing complete");
        }


        executor = Executors.newFixedThreadPool(5);
        listeners = new CopyOnWriteArrayList<ListenerContainer>();
        log.info("Storage <{}> created successfully", name);
    }

    public void postMessage(final Message message) {
        executor.execute(new Runnable() {
            public void run() {
                TokenizedMessage tokenizedMessage = new TokenizedMessage(message);
                for (ListenerContainer listener : listeners) {
                    listener.processMessage(tokenizedMessage);
                }
            }
        });

        executor.execute(new Runnable() {
            public void run() {
                try {
                    PreparedStatement storeStatement = connection.prepareStatement(INSERT_MESSAGE_QUERY);
                    storeStatement.setObject(1, message.getId());
                    storeStatement.setLong(2, message.getTime());
                    storeStatement.setString(3, message.getAuthor());
                    storeStatement.setString(4, message.getText());
                    storeStatement.executeUpdate();
                } catch (SQLException e) {
                    if (e.getErrorCode() == DUPLICATE_UUID_SQL_ERROR_CODE) {
                        log.trace("Attempt to write message that already in database. Message: {}", message);
                    } else {
                        log.error("Error writing message to database", e);
                    }
                }
            }
        });
    }

    public MessageQuery createQuery(final QueryRequest request) {
        try {
            return new MessageQueryImpl(request, new DatabaseBatchQuery(request, connection), listeners);
        } catch (SQLException e) {
            log.error("Can't create DatabaseBatchQuery instance", e);
            throw new RuntimeException(e);
        }
    }

}
